package com.zdp.contentcenter.rocketmq.tx;

import com.zdp.contentcenter.dao.content.RocketmqTransactionLogMapper;
import com.zdp.contentcenter.domain.dto.content.ShareAuditDTO;
import com.zdp.contentcenter.domain.entity.content.RocketmqTransactionLog;
import com.zdp.contentcenter.service.content.ShareService;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;

/**
 * @author sesshomaru
 * @date 2020/9/23 21:10
 *
 * 监听事务消息
 */
// txProducerGroup一定要与发送消息的group一致
@RocketMQTransactionListener(txProducerGroup = "tx-add-bonus-group")
@RequiredArgsConstructor(onConstructor = @_(@Autowired))
public class AddBonusTransactionListener implements RocketMQLocalTransactionListener {

    private final ShareService shareService;

    private final RocketmqTransactionLogMapper rocketmqTransactionLogMapper;

    // 这个用来执行本地事务的
    // 获取事务消息中传递的内容
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        Integer shareId = Integer.valueOf((String) headers.get("share_id"));

        try {
            // 执行本地事务同时记录日志
            shareService.auditByIdWithRocketMqLog(shareId, (ShareAuditDTO) arg, transactionId);
            // 这里测试需要直接kill进程的方式阻止确认消息发出,idea的停止时优雅停止它会把代码执行完后再停下来
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    // 这个是回查方法，如果半消息(事务消息)的状态一直没有改变，rocketmq就会自动去调用这个方法去回查本地事务的情况（这里采用日志的方法记录本地事务的情况）
    // 有日志则表示本地事务执行成功 没有事务则表示不成功，因为本消息携带了这个消息的事务id，而executeLocalTransaction这个本地事务和记录日志绑定为一个事务
    // 也就是说本地事务执行成功那么这个消息的事务id就会存在与日志记录表中，即可通过日志表判断本地事务是否success
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = (String) headers.get(RocketMQHeaders.TRANSACTION_ID);
        // select * from xxx where transaction_id = xxx
        RocketmqTransactionLog rocketmqTransactionLog = rocketmqTransactionLogMapper.selectOne(
                RocketmqTransactionLog.builder()
                        .transactionId(transactionId)
                        .build()
        );
        if (rocketmqTransactionLog != null) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}
